package com.shujia.flink.source

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.source.{RichParallelSourceFunction, RichSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala._

object Demo5MysqlSource {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(4)

    val mysqlDS: DataStream[(String, String, Int, String, String)] = env.addSource(new MysqlSource)

    mysqlDS.print()

    env.execute()

  }

}

/**
  * SourceFunction: 基础的SourceFunction, 单线程的source
  * RichSourceFunction: 比SourceFunction多了open和close方法 -单线程的source
  * ParallelSourceFunction ： 多并行的source
  * RichParallelSourceFunction 比SourceFunction多了open和close方法  -多并行的source
  */

class MysqlSource extends RichSourceFunction[(String, String, Int, String, String)] {

  /**
    * open: 在run方法之前执行，一般用于初始化，比如创建网络链接
    */

  var con: Connection = _

  override def open(parameters: Configuration): Unit = {
    println("创建链接")

    Class.forName("com.mysql.jdbc.Driver")

    //建立链接
    con = DriverManager.getConnection("jdbc:mysql://master:3306/bigdata", "root", "123456")

  }

  /**
    * close 在run方法执行执行，一般用于回收资源
    *
    */
  override def close(): Unit = {
    println("关闭链接")
    //关闭链接
    con.close()
  }

  /**
    * 再run方法中读取数据， 如果run会结束-- 有界流  ，如果run不会结束 --无界流
    *
    */

  override def run(ctx: SourceFunction.SourceContext[(String, String, Int, String, String)]): Unit = {

    println("run")


    //使用jdbc读取mysq

    //查询数据
    val stat: PreparedStatement = con.prepareStatement("select * from student")

    //执行查询
    val resultSet: ResultSet = stat.executeQuery()

    //循环解析数据
    while (resultSet.next()) {
      val id: String = resultSet.getString("id")
      val name: String = resultSet.getString("name")
      val age: Int = resultSet.getInt("age")
      val gender: String = resultSet.getString("gender")
      val clazz: String = resultSet.getString("clazz")


      //将数据发送到下游
      ctx.collect((id, name, age, gender, clazz))
    }


  }

  override def cancel(): Unit = {

  }
}
